#include "mysql.hh"
#include "../thirdparty/mysql/internal/mysql_row.h"
#include "../thirdparty/mysql/internal/mysql_connection.h"
#include "../thirdparty/mysql/internal/mysql_resultset.h"
#include "../object_pool.hpp"
#include "../logger/writer.hh"
#include <stdexcept>
#include <sstream>
#include <random>

#ifdef KRATOS_COMPILER_LOG_OFF

#define MYSQL_LOG_ERROR(a, b)

#define MYSQL_LOG_EXCEPTION(e, a, b)

#define MYSQL_PARAM_FAIL_RETURN_FALSE(a)\
    return false;

#define MYSQL_PARAM_FAIL_RETURN(a, b)\
    return b;

#else

#define MYSQL_LOG_ERROR(a, b)\
    logwcom("MySQL") << a << b << corelib::logend;

#define MYSQL_LOG_EXCEPTION(e, a, b)\
    logwcom("MySQL") << "Exception:" << e.what() <<","<< a << b << corelib::logend;

#define MYSQL_PARAM_FAIL_RETURN_FALSE(a)\
    if (a) {\
        logwcom("MySQL") << "[" << __FILE__ << ":" << __LINE__ << "]:" << #a << ", check failed" << corelib::logend;\
        return false;\
    }

#define MYSQL_PARAM_FAIL_RETURN(a, b)\
    if (a) {\
        logwcom("MySQL") << "[" << __FILE__ << ":" << __LINE__ << "]:" << #a << ", check failed" << corelib::logend;\
        return b;\
    }

#endif // KRATOS_COMPILER_LOG_OFF

namespace kratos { namespace mysql {

MySQLResultRow::MySQLResultRow(krdb::MysqlRow * row, std::size_t fieldCount) {
    auto mysql_row = row->getRaw();
    auto lengths = row->getLengths();
    for (std::size_t i = 0; i < fieldCount; i++) {
        fields_.emplace_back(std::string(mysql_row[i], lengths[i]));
    }
}

MySQLResultRow::~MySQLResultRow() {
}

const StringVector & MySQLResultRow::getFields() const {
    return fields_;
}

MySQLResult::MySQLResult(std::uint32_t sequence, std::uint32_t taskID, std::uint64_t userData, std::string& command)
    : errorID_(-1), sequence_(sequence), taskID_(taskID), userData_(userData), command_(command) {
}

MySQLResult::~MySQLResult() {
    for (auto& row : rows_) {
        corelib::dispose(row);
    }
    rows_.clear();
}

const std::string & MySQLResult::getErrorString() const {
    return error_;
}

int MySQLResult::getErrorID() const {
    return errorID_;
}

MySQLResult::operator bool() {
    return (errorID_ != -1);
}

const FieldNames & MySQLResult::getFields() const {
    return fields_;
}

const Rows & MySQLResult::getRows() const {
    return rows_;
}

void MySQLResult::setErrorString(const std::string& error) {
    error_ = error;
}

void MySQLResult::setErrorID(int errorID) {
    errorID_ = errorID;
}

void MySQLResult::addFields(char** name, std::size_t fieldCount) {
    for (std::size_t i = 0; i < fieldCount; i++) {
        fields_.push_back(name[i]);
    }    
}

void MySQLResult::addRow(krdb::RdbRow * row, std::size_t fieldCount) {
    auto mysqlRow = corelib::allocate<MySQLResultRow>(dynamic_cast<krdb::MysqlRow*>(row), fieldCount);
    if (!mysqlRow) {
        MYSQL_LOG_ERROR("cannot fork MySQLResultRow ",
            "corelib::allocate<MySQLResultRow>(dynamic_cast<krdb::MysqlRow*>(row), fieldCount)");
        return;
    }
    rows_.push_back(mysqlRow);
}

std::uint32_t MySQLResult::getSequence() {
    return sequence_;
}

std::uint32_t MySQLResult::getTaskID() {
    return taskID_;
}

std::uint64_t MySQLResult::getUserData() {
    return userData_;
}

const std::string & MySQLResult::getCommand() const {
    return command_;
}

MySQLConn::MySQLConn(const std::string& host, int port, const std::string& user, const std::string& password,
    const std::string& db)
    : host_(host), port_(port), user_(user), password_(password), database_(db), conn_(nullptr), run_(false) {
}

MySQLConn::~MySQLConn() {
    stop();
}

void MySQLConn::update(std::time_t now) {
    if (!run_) {
        return;
    }
    MySQLResult* result = nullptr;
    while (output_.try_dequeue(result)) {
        auto taskIt = taskMap_.find(result->getSequence());
        if (taskIt != taskMap_.end()) {
            auto& info = taskIt->second;
            if (info.cb) {
                try {
                    info.cb(*result);
                } catch (std::exception& e) {
                    MYSQL_LOG_EXCEPTION(e, "Process result for command:", info.command);
                }
                if (info.timerID != corelib::INVALID_TIMER_ID) {
                    timerQueue_.cancel(info.timerID);
                }
            }
            taskMap_.erase(taskIt);
        }
        corelib::dispose(result);
    }
    auto timerID = timerQueue_.checkExpired();
    while (timerID != corelib::INVALID_TIMER_ID) {
        auto timerIt = timerMap_.find(timerID);
        if (timerIt != timerMap_.end()) {
            auto taskIt = taskMap_.find(timerIt->second);
            if (taskIt != taskMap_.end()) {
                auto& info = taskIt->second;
                if (info.tocb) {
                    try {
                        info.tocb(info.command, info.taskID, info.userData);
                    } catch (std::exception& e) {
                        MYSQL_LOG_EXCEPTION(e, "Timeout command:", info.command);
                    }
                } else {
                    MYSQL_LOG_ERROR("Timeout command:", info.command);
                }
                taskMap_.erase(taskIt);
            }
            timerMap_.erase(timerIt);
        }        
        timerID = timerQueue_.checkExpired();
    }
}

bool MySQLConn::start() {
    MYSQL_PARAM_FAIL_RETURN_FALSE(run_);
    conn_ = corelib::allocate<krdb::MysqlConnection>();
    MYSQL_PARAM_FAIL_RETURN_FALSE(!conn_);
    try {
        conn_->connect(host_, port_, user_, password_, database_);
    } catch (std::exception& e) {
        MYSQL_LOG_EXCEPTION(e, "When connect to ", *this);
        corelib::dispose(conn_);
        conn_ = nullptr;
        return false;
    }
    run_ = true;
    routine_.reset(new std::thread([&, this] {
        while (run_) {
            MySQLResult* result = nullptr;
            while (input_.try_dequeue(result)) {
                // do command and get result
                krdb::RdbResult* mysql_result = nullptr;
                bool except = false;
                try {
                    mysql_result = conn_->execute(conn_->escape(result->getCommand()));
                } catch (std::exception& e) {
                    except = true;
                    result->setErrorID(-1);
                    result->setErrorString(std::string("Exception:") + e.what());
                }
                if (!except) {                    
                    result->setErrorID(mysql_result->getErrorId());
                    result->setErrorString(mysql_result->getErrorString());
                    if (mysql_result->hasResultSet()) {
                        auto resultset = mysql_result->getResultSet();
                        auto fieldNum = resultset->getFieldNum();
                        result->addFields(resultset->getFieldNames(), fieldNum);
                        while (resultset->hasNext()) {
                            auto next = resultset->next();
                            result->addRow(next, fieldNum);
                        }
                    }
                }
                if (!output_.try_enqueue(result)) {
                    MYSQL_LOG_ERROR("result output queue error", "output_.try_enqueue(result)");
                    corelib::dispose(result);
                    continue;
                }
            }
            std::this_thread::sleep_for(std::chrono::milliseconds(1));
        }
        corelib::dispose(conn_);
        conn_ = nullptr;
        MySQLResult* result = nullptr;
        while (input_.try_dequeue(result)) {
            corelib::dispose(result);
        }        
    }));
    return true;
}

void MySQLConn::stop() {
    if (!run_) {
        return;
    }
    run_ = false;
    if (routine_) {
        routine_->join();
    }
    MySQLResult* result = nullptr;
    while (output_.try_dequeue(result)) {
        corelib::dispose(result);
    }
}

bool MySQLConn::doCommand(const std::string & SQLCommand, std::uint32_t taskID, std::uint64_t userData, MySQLCallback cb,
    MySQLTimeoutCallback tocb, std::time_t timeout) {
    MYSQL_PARAM_FAIL_RETURN_FALSE(!run_);
    MYSQL_PARAM_FAIL_RETURN_FALSE(SQLCommand.empty());
    static std::uint32_t sequence = 1;
    corelib::TimerID timerID = corelib::INVALID_TIMER_ID;
    if (!timeout) {
        timeout = 5000;
    }
    timerID = timerQueue_.startTimerOnce(timeout, 0);
    MYSQL_PARAM_FAIL_RETURN_FALSE(timerID == corelib::INVALID_TIMER_ID);
    auto info = TaskInfo{SQLCommand, taskID, userData, timerID, cb, tocb};
    auto newSequence = sequence++;
    auto result = corelib::allocate<MySQLResult>(newSequence, taskID, userData, SQLCommand);
    MYSQL_PARAM_FAIL_RETURN_FALSE(!result);
    taskMap_.emplace(newSequence, info);
    timerMap_[timerID] = newSequence;
    if (!input_.try_enqueue(result)) {
        timerQueue_.cancel(timerID);
        taskMap_.erase(newSequence);
        timerMap_.erase(timerID);
        corelib::dispose(result);
        return false;
    }
    return true;
}

MySQLConn::operator std::string() {
    std::stringstream ss;
    ss << "Host:" << host_
       << ", Port:" << port_
       << ", User:" << user_
       << ", Password:" << password_
       << ", Database:" << database_;
    return ss.str();
}

MySQL::MySQL() {
}

MySQL::~MySQL() {
    stop();
}

void MySQL::stop() {
    for (auto& it : connMap_) {
        for (auto conn : it.second) {
            conn->stop();
            corelib::dispose(conn);
        }
    }
    connMap_.clear();
}

bool MySQL::addHost(const std::string & name, const std::string & host, int port, const std::string & user,
    const std::string & password, const std::string & db, std::size_t maxPoolSize) {
    MYSQL_PARAM_FAIL_RETURN_FALSE(name.empty());
    MYSQL_PARAM_FAIL_RETURN_FALSE(host.empty());
    if (!maxPoolSize) {
        maxPoolSize = 8;
    }
    auto& conns = connMap_[name];
    for (std::size_t i = 0; i < maxPoolSize; i++) {
        auto conn = corelib::allocate<MySQLConn>(host, port, user, password, db);
        MYSQL_PARAM_FAIL_RETURN_FALSE(!conn);
        MYSQL_PARAM_FAIL_RETURN_FALSE(!conn->start());
        conns.push_back(conn);
    }
    return true;
}

MySQLConn * MySQL::findConn(const std::string & name, MySQLHostFindPolicy /*policy*/) {
    MYSQL_PARAM_FAIL_RETURN(name.empty(), nullptr);
    static std::default_random_engine e;
    auto it = connMap_.find(name);
    if (it == connMap_.end()) {
        return nullptr;
    }
    auto& conns = it->second;
    if (conns.empty()) {
        return nullptr;
    }
    return conns[e() % conns.size()];
}

void MySQL::update(std::time_t now) {
    for (auto& it : connMap_) {
        for (auto conn : it.second) {
            conn->update(now);
        }
    }
}

}}
